/**
 * Illustrates a wordcount in Java
 */
package com.hsj;


import java.util.*;
import java.lang.Iterable;

import org.apache.spark.SparkConf;
import org.apache.spark.rdd.RDD;
import scala.Tuple2;

import org.apache.commons.lang.StringUtils;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;

/**
 * 单词统计样例
 */
public class WordCount {
    public static void main(String[] args) throws Exception {
        String master = "local";
//		JavaSparkContext sc = new JavaSparkContext(
//      master, "wordcount", System.getenv("SPARK_HOME"), System.getenv("JAVA_HOME"));


        SparkConf conf = new SparkConf().setMaster("local").setAppName("WordCount");
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaRDD<String> rdd = sc.textFile("C:\\dev\\hadoop-2.7.7\\README.txt");

        //单词统计
        JavaPairRDD<String, Integer> counts = rdd.flatMap(
                new FlatMapFunction<String, String>() {
                    public Iterator<String> call(String x) {
                        return Arrays.asList(x.split(" ")).iterator();
                    }
                }).mapToPair(new PairFunction<String, String, Integer>() {
            public Tuple2<String, Integer> call(String x) {
                return new Tuple2(x, 1);
            }
        }).reduceByKey(new Function2<Integer, Integer, Integer>() {
            public Integer call(Integer x, Integer y) {
                return x + y;
            }
        });

        //更简洁的单词统计
        Map<String, Long> counts2 = rdd.flatMap(new FlatMapFunction<String, String>() {
            public Iterator<String> call(String x) {
                return Arrays.asList(x.split(" ")).iterator();
            }
        }).countByValue();

        long num = counts.count();
//    System.out.println(counts.collect());

        counts.saveAsTextFile("WordCountResult");


    }
}
